package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"

	"bytes"
	"math/rand"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"

	"6.824/labgob"
	"6.824/labrpc"
)

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	// For Lab3
	CommandTerm int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

var HEARTBEAT_PERIOD time.Duration = 200 * time.Millisecond
var LOST_HEARTBEAT_PERIOD time.Duration = 2*HEARTBEAT_PERIOD + 100*time.Millisecond
var ELECTION_TIMEOUT_PERIOD time.Duration = 200 * time.Millisecond

type LogEntry struct {
	Index int
	Cmd   interface{}
	Term  int
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	currentTerm int
	votedFor    int
	log         []LogEntry

	commitIndex int
	lastApplied int
	nextIndex   []int
	matchIndex  []int

	// 辅助变量
	applyCh        chan ApplyMsg
	actionTimer    time.Timer
	role           int // 0:Leader, 1:Follower, 2:Candicater
	applyLogSignal chan bool
}

// for 2D
type InstallSnapshotArgs struct {
	Term              int
	LeaderId          int
	LastIncludedIndex int
	LastIncludedTerm  int
	Offset            int
	Data              []byte
	Done              bool
}

type InstallSnapshotReply struct {
	Term int
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	var term int
	var isleader bool
	// Your code here (2A).
	term = rf.currentTerm
	isleader = rf.role == 0
	return term, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// data := w.Bytes()
	// rf.persister.SaveRaftState(data)

	rf.persister.SaveRaftState(rf.persistData())
}

func (rf *Raft) persistData() []byte {

	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)

	e.Encode(rf.currentTerm)
	e.Encode(rf.votedFor)
	e.Encode(rf.log)

	data := w.Bytes()
	return data
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }

	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)

	var currentTerm int
	var votedFor int
	var log []LogEntry

	d.Decode(&currentTerm)
	d.Decode(&votedFor)
	d.Decode(&log)

	rf.currentTerm = currentTerm
	rf.votedFor = votedFor
	rf.log = log
	rf.lastApplied = log[0].Index

}

// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

func (rf *Raft) applySnapShot(msg ApplyMsg, args *InstallSnapshotArgs) {
	rf.mu.Lock()
	clearChannel(rf.applyLogSignal)
	rf.log = []LogEntry{{args.LastIncludedIndex, -100, args.LastIncludedTerm}}
	rf.commitIndex = args.LastIncludedIndex
	rf.lastApplied = rf.commitIndex
	rf.SnapshotNotLock(args.LastIncludedIndex, args.Data)
	rf.persist()
	rf.mu.Unlock()
	rf.applyCh <- msg
}

func (rf *Raft) InstallSnapshot(args *InstallSnapshotArgs, reply *InstallSnapshotReply) {
	// Your code here (2D).

	rf.mu.Lock()

	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		rf.mu.Unlock()
		return
	} else {
		rf.currentTerm = args.Term
		reply.Term = rf.currentTerm
	}

	rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
	rf.role = 1
	rf.votedFor = -1

	rf.mu.Unlock()

	msg := ApplyMsg{
		SnapshotValid: true,
		Snapshot:      args.Data,
		SnapshotIndex: args.LastIncludedIndex,
		SnapshotTerm:  args.LastIncludedTerm,
	}
	rf.applySnapShot(msg, args)

}

// For Lab3B
func (rf *Raft) ZeroLogIndex() int {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	return rf.log[0].Index
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock() // 注意在applyCh处解锁
	remain_num := rf.log[len(rf.log)-1].Index - index

	zero_log := rf.log[0]
	if index < zero_log.Index || remain_num < 0 {
		rf.mu.Unlock()
		return
	}

	Assert(remain_num >= 0, "not remain_num>=0")
	right_rest := rf.log[rf.log[len(rf.log)-1].Index+1-remain_num-zero_log.Index : rf.log[len(rf.log)-1].Index+1-zero_log.Index]
	zero_log.Term = rf.log[index-zero_log.Index].Term
	rf.log = make([]LogEntry, remain_num+1)
	zero_log.Index = index // 用于标记已存入快照的最高索引
	rf.log[0] = zero_log
	copy(rf.log[1:], right_rest)

	data := rf.persistData()
	rf.persister.SaveStateAndSnapshot(data, snapshot)
	rf.mu.Unlock()

}

func (rf *Raft) SnapshotNotLock(index int, snapshot []byte) {
	// Your code here (2D).
	zero_log := rf.log[0]
	remain_num := rf.log[len(rf.log)-1].Index - index
	right_rest := rf.log[rf.log[len(rf.log)-1].Index+1-remain_num-zero_log.Index : rf.log[len(rf.log)-1].Index+1-zero_log.Index]
	zero_log.Term = rf.log[index-zero_log.Index].Term
	rf.log = make([]LogEntry, remain_num+1)
	zero_log.Index = index // 用于标记已存入快照的最高索引
	rf.log[0] = zero_log
	copy(rf.log[1:], right_rest)

	data := rf.persistData()
	rf.persister.SaveStateAndSnapshot(data, snapshot)
}

func (rf *Raft) SnapshotForServer() []byte {
	return rf.persister.ReadSnapshot()
}

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

type AppendEntriesArgs struct {
	Term         int // 发送消息的leader的当前任期
	LeaderId     int
	PrevLogIndex int        // 新日志指令项之前的指令项索引
	PrevLogTerm  int        // 新日志指令项之前的指令项任期
	Entries      []LogEntry // 发送的指令项
	LeaderCommit int        // leader已提交的索引号
}

type AppendEntriesReply struct {
	Term    int
	Success bool
}

func (rf *Raft) logIndexAbsoluted2Relative(absolute int) int {
	return absolute - rf.log[0].Index
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if args.Term < rf.currentTerm {
		// println("fail vote expire term:" + strconv.Itoa(args.CandidateId) + "->" + strconv.Itoa(rf.me))
		reply.VoteGranted = false
		reply.Term = rf.currentTerm
		return
	}

	// 避免产生相同（任期，索引）的日志
	if args.Term == rf.currentTerm && rf.votedFor != args.CandidateId {
		reply.VoteGranted = false
		reply.Term = rf.currentTerm
		return
	}

	if args.Term > rf.currentTerm {
		rf.role = 1
		rf.votedFor = -1
		rf.currentTerm = args.Term
	}

	reply.Term = rf.currentTerm

	// more up-to-date
	if (rf.votedFor == -1 || rf.votedFor == args.CandidateId) && (args.LastLogTerm > rf.log[len(rf.log)-1].Term ||
		(args.LastLogTerm == rf.log[len(rf.log)-1].Term && args.LastLogIndex >= rf.log[len(rf.log)-1].Index)) {
		// println("success vote:" + strconv.Itoa(args.CandidateId) + "->" + strconv.Itoa(rf.me) + " term:" + strconv.Itoa(args.Term))
		rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
		reply.VoteGranted = true
		rf.votedFor = args.CandidateId
		rf.role = 1
	} else {
		reply.VoteGranted = false
		// println("fail vote invalid:" + strconv.Itoa(rf.me) + "->" +
		// 	strconv.Itoa(args.CandidateId) + " : " + strconv.Itoa(args.LastLogTerm) +
		// 	"-" + strconv.Itoa(rf.log[len(rf.log)-1].Term) + " " + strconv.Itoa(args.LastLogIndex) + "-" + strconv.Itoa(rf.log[len(rf.log)-1].Index))
	}

	rf.persist()

}

// 同步日志
func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	if args.Term < rf.currentTerm {
		reply.Term = rf.currentTerm
		reply.Success = false
		rf.mu.Unlock()
		return
	}

	// If AppendEntries RPC received from new leader: convert to follower
	// transfer follower
	rf.currentTerm = args.Term
	rf.role = 1
	rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
	rf.votedFor = -1

	// 落后
	if args.PrevLogIndex > rf.log[len(rf.log)-1].Index {
		// println(strconv.Itoa(rf.me) + " behind " + strconv.Itoa(args.PrevLogIndex) + ", " + strconv.Itoa(rf.log[len(rf.log)-1].Index))
		reply.Success = false
		rf.persist()
		rf.mu.Unlock()
		return
	}

	// 检查任期是否有问题
	if rf.logIndexAbsoluted2Relative(args.PrevLogIndex) >= 0 &&
		rf.log[rf.logIndexAbsoluted2Relative(args.PrevLogIndex)].Term != args.PrevLogTerm {
		// println(strconv.Itoa(rf.me) + " conflict " + strconv.Itoa(args.PrevLogIndex) + " " + strconv.Itoa(rf.log[rf.logIndexAbsoluted2Relative(args.PrevLogIndex)].Term) + ", " + strconv.Itoa(args.PrevLogTerm))
		reply.Success = false
		rf.persist()
		rf.mu.Unlock()
		return
	}

	// 注意： 获取锁的时机是随机的，如果旧RPC到达比新RPC晚，不当的操作会导致指令丢失
	// 遍历新来的指令项，冲突或插入
	for i := 0; i < len(args.Entries); i++ {
		if args.PrevLogIndex+i+1 != args.Entries[i].Index {
			panic("args.PrevLogIndex+i != args.Entries[i].Index" + "(" + strconv.Itoa(args.PrevLogIndex+i+1) + ", " + strconv.Itoa(args.Entries[i].Index) + ")")
		}

		if rf.logIndexAbsoluted2Relative(args.Entries[i].Index) < 1 {
			continue
		}

		if rf.logIndexAbsoluted2Relative(args.Entries[i].Index) < len(rf.log) &&
			rf.log[rf.logIndexAbsoluted2Relative(args.Entries[i].Index)].Term != args.Entries[i].Term {

			// 如果之前指令的任期不同当前leader指令的任期，则认为冲突了，此时应该以leader为准，覆盖自身的指令
			// conflict, replace
			rf.log = rf.log[0 : rf.logIndexAbsoluted2Relative(args.Entries[i].Index)+1] // 如果出现一个冲突的，则后续指令应全部去除
			rf.log[rf.logIndexAbsoluted2Relative(args.Entries[i].Index)] = args.Entries[i]
			// println(strconv.Itoa(rf.me) + " conflict " + strconv.Itoa(args.Entries[i].Index) + " : " + strconv.Itoa(rf.log[args.Entries[i].Index].Cmd.(int)))
		} else if rf.logIndexAbsoluted2Relative(args.Entries[i].Index) >= len(rf.log) {
			if rf.logIndexAbsoluted2Relative(args.Entries[i].Index) != len(rf.log) {
				panic("rf.logIndexAbsoluted2Relative(args.Entries[i].Index)!=len(rf.log)")
			}
			// append new entry
			rf.log = append(rf.log, args.Entries[i])
			// println(strconv.Itoa(rf.me) + " append new " + strconv.Itoa(args.Entries[i].Index) + " : " + strconv.Itoa(args.Entries[i].Cmd.(int)))
		} else {
			// println(strconv.Itoa(args.LeaderId) + "->" + strconv.Itoa(rf.me) + " invalid entry:" + strconv.Itoa(args.Entries[i].Index) + " -- " + strconv.Itoa(rf.log[len(rf.log)-1].Index+1))
		}
	}

	// 按理说不应该出现log比commitIndex小的
	if rf.log[len(rf.log)-1].Index < rf.commitIndex {
		panic("rf.log[len(rf.log)-1].Index")
	}

	lastCommit := rf.commitIndex
	rf.commitIndex = args.LeaderCommit
	if rf.log[len(rf.log)-1].Index < args.LeaderCommit {
		rf.commitIndex = rf.log[len(rf.log)-1].Index
	}
	if rf.commitIndex < lastCommit {
		rf.commitIndex = lastCommit
	}

	shouldApply := rf.lastApplied < rf.commitIndex

	reply.Success = true
	rf.persist()
	rf.mu.Unlock()

	if shouldApply {
		rf.applyLogSignal <- true
	}
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

func (rf *Raft) sendInstallSnapshot(server int, args *InstallSnapshotArgs, reply *InstallSnapshotReply) bool {
	ok := rf.peers[server].Call("Raft.InstallSnapshot", args, reply)
	return ok
}

// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	index := rf.log[len(rf.log)-1].Index + 1
	term := rf.currentTerm
	isLeader := rf.role == 0

	// Your code here (2B).
	if isLeader {
		// println(strconv.Itoa(rf.me) + " ---------- new cmd: index=" + strconv.Itoa(index) + " term=" + strconv.Itoa(term))
		rf.log = append(rf.log, LogEntry{index, command, term})
		rf.actionTimer.Reset(HEARTBEAT_PERIOD)
		rf.persist()
		rf.mu.Unlock()
		rf.sync_appendEntries_paralleled()
	} else {
		rf.mu.Unlock()
	}

	return index, term, isLeader
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

// for 2D
func (rf *Raft) send_installsnapshot(s int) bool {
	for !rf.killed() {
		rf.mu.Lock()
		if rf.role != 0 {
			rf.mu.Unlock()
			return false
		}
		last_index, last_term := rf.log[0].Index, rf.log[0].Term
		req := InstallSnapshotArgs{rf.currentTerm, rf.me, last_index, last_term, 0, rf.persister.ReadSnapshot(), true}
		reply := InstallSnapshotReply{0}
		rf.mu.Unlock()
		ok := rf.sendInstallSnapshot(s, &req, &reply)

		if ok {
			rf.mu.Lock()
			if rf.currentTerm < reply.Term {
				rf.role = 1
				rf.votedFor = -1
				rf.currentTerm = reply.Term
				rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
				rf.persist()
				rf.mu.Unlock()
				return false
			}

			if last_index > rf.matchIndex[s] {
				rf.nextIndex[s] = last_index + 1
				rf.matchIndex[s] = last_index
			}

			rf.mu.Unlock()
			return true
		}

		time.Sleep(10 * time.Millisecond)
	}

	return false
}

func (rf *Raft) timeout_voted_one(s int, req *RequestVoteArgs, vote_count *int) {
	for !rf.killed() {
		rf.mu.Lock()
		if rf.role != 2 {
			rf.mu.Unlock()
			return
		}
		rf.mu.Unlock()
		reply := RequestVoteReply{-1, false}
		ok := rf.sendRequestVote(s, req, &reply)

		if ok {
			rf.mu.Lock()
			if reply.Term > rf.currentTerm {
				// transfer follower
				rf.role = 1
				rf.votedFor = -1
				rf.currentTerm = reply.Term
				rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
				rf.persist()
				rf.mu.Unlock()
				return
			}

			if reply.VoteGranted {
				(*vote_count)++

				if *vote_count >= len(rf.peers)/2+1 && rf.role == 2 {
					// transfer leader
					rf.role = 0
					rf.votedFor = -1
					rf.actionTimer.Reset(50 * time.Millisecond) // Lab2a:立即发送心跳;Lab2b:延迟一段时间发送心跳, 确认是否为Leader
					// println("%%%%%%%%%%%%%% " + strconv.Itoa(rf.me) + " transfer leader term:" + strconv.Itoa(rf.currentTerm) + " log.len=" + strconv.Itoa(len(rf.log)))
					// 设置待各随从的待同步服务日志索引
					for i := 0; i < len(rf.peers); i++ {
						rf.nextIndex[i] = rf.log[len(rf.log)-1].Index + 1
						rf.matchIndex[i] = 0
					}
					rf.persist()
				}
			}
			rf.mu.Unlock()
			return
		}

		time.Sleep(10 * time.Millisecond)
	}
}

// 并行请求投票
func (rf *Raft) start_vote_paralleled() {
	vote_count := 1 // me first
	rf.mu.Lock()
	if rf.role != 2 {
		rf.mu.Unlock()
		return
	}
	req := RequestVoteArgs{rf.currentTerm, rf.me, rf.log[len(rf.log)-1].Index, rf.log[len(rf.log)-1].Term}
	rf.mu.Unlock()
	// println(time.Now().String() + "===" + strconv.Itoa(rf.me) + " start vote term:" + strconv.Itoa(rf.currentTerm))
	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}

		go rf.timeout_voted_one(i, &req, &vote_count)
	}
}

// 找出matchIndex里的N
func (rf *Raft) find_matchIndex_N() int {
	ret := rf.commitIndex
	left, right := rf.commitIndex, rf.log[len(rf.log)-1].Index
	for left <= right {
		mid := (left + right) / 2
		count := 1
		for i := 0; i < len(rf.peers); i++ {
			// rf.matchIndex[me] is always 0, so...
			if rf.matchIndex[i] >= mid {
				count++
			}
		}

		if count >= len(rf.peers)/2+1 {
			ret = mid
			left = mid + 1
		} else {
			right = mid - 1
		}
	}

	return ret
}

func clearChannel(ch chan bool) {
	for {
		select {
		case <-ch:

		default:
			return
		}
	}
}

// 日志提交的守护线程
func (rf *Raft) applyLogDaemon() {
	for !rf.killed() {
		select {
		case <-rf.applyLogSignal:
			{
				clearChannel(rf.applyLogSignal)
				rf.mu.Lock()

				if rf.role == 0 {
					N := rf.find_matchIndex_N()
					if rf.logIndexAbsoluted2Relative(N) > 0 && rf.log[rf.logIndexAbsoluted2Relative(N)].Term == rf.currentTerm {
						rf.commitIndex = N
					}
				}

				for rf.lastApplied < rf.commitIndex {
					rf.lastApplied++
					toAppliedIndex := rf.lastApplied
					Assert(rf.logIndexAbsoluted2Relative(toAppliedIndex) > 0, "rf.lastApplied not right :"+strconv.Itoa(rf.logIndexAbsoluted2Relative(toAppliedIndex)))
					toAppliedLog := rf.log[rf.logIndexAbsoluted2Relative(toAppliedIndex)]
					// println(strconv.Itoa(rf.me) + " commmit " + strconv.Itoa(rf.lastApplied))
					Assert(rf.lastApplied == rf.log[rf.logIndexAbsoluted2Relative(rf.lastApplied)].Index, "not equal")
					rf.mu.Unlock()
					rf.applyCh <- ApplyMsg{CommandValid: true, CommandIndex: toAppliedIndex, Command: toAppliedLog.Cmd, CommandTerm: toAppliedLog.Term}
					rf.mu.Lock()
				}
				rf.mu.Unlock()
			}
		}
	}
}

// 向终端s同步指令日志
func (rf *Raft) sync_appendEntries(s int) {
	for !rf.killed() {
		reply := AppendEntriesReply{-1, false}
		send_entries := []LogEntry{}

		rf.mu.Lock()
		if rf.role != 0 {
			rf.mu.Unlock()
			return
		}

		if rf.nextIndex[s] == -1 {
			rf.mu.Unlock()
			return
		}

		if rf.nextIndex[s] < rf.log[0].Index+1 {
			rf.nextIndex[s] = rf.log[0].Index + 1
		}

		for i := rf.nextIndex[s]; i < rf.log[len(rf.log)-1].Index+1; i++ {
			send_entries = append(send_entries, rf.log[rf.logIndexAbsoluted2Relative(i)])
		}

		lastLogIndex := rf.log[rf.logIndexAbsoluted2Relative(rf.nextIndex[s]-1)].Index
		lastLogTerm := rf.log[rf.logIndexAbsoluted2Relative(rf.nextIndex[s]-1)].Term
		currentTopIndex := rf.log[len(rf.log)-1].Index

		send_req := AppendEntriesArgs{rf.currentTerm, rf.me, lastLogIndex, lastLogTerm, send_entries, rf.commitIndex}
		rf.mu.Unlock()

		ok := rf.sendAppendEntries(s, &send_req, &reply)

		if ok {
			rf.mu.Lock()
			if rf.role != 0 {
				rf.mu.Unlock()
				return
			}

			if reply.Term > rf.currentTerm {
				// transfer follower
				rf.role = 1
				rf.votedFor = -1
				rf.currentTerm = reply.Term
				rf.actionTimer.Reset(LOST_HEARTBEAT_PERIOD)
				rf.persist()
				rf.mu.Unlock()
				return
			}

			if reply.Success {
				if currentTopIndex > rf.matchIndex[s] {
					rf.matchIndex[s] = currentTopIndex
				}

				if currentTopIndex+1 > rf.nextIndex[s] {
					rf.nextIndex[s] = currentTopIndex + 1
				}

				rf.mu.Unlock()

				rf.applyLogSignal <- true
				return

			} else {
				toDecendNextIndex := lastLogIndex - 500
				// 如果toDecendNextIndex >= rf.nextIndex[s], 则意味着已有其他线程进行以下操作
				if toDecendNextIndex < rf.nextIndex[s] {
					if toDecendNextIndex < rf.matchIndex[s]+1 {
						rf.nextIndex[s] = rf.matchIndex[s] + 1
					} else {
						rf.nextIndex[s] = toDecendNextIndex
					}

					if rf.nextIndex[s] < rf.log[0].Index+1 {
						rf.nextIndex[s] = -1
						rf.mu.Unlock()
						send_res := rf.send_installsnapshot(s)
						if !send_res {
							return
						}
					} else {
						rf.mu.Unlock()
					}
				} else {
					rf.mu.Unlock()
				}

			}

		}

		time.Sleep(10 * time.Millisecond)

	}

}

// 并行同步应用日志并获取请求
func (rf *Raft) sync_appendEntries_paralleled() {
	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}

		go rf.sync_appendEntries(i)
	}
}

// 并行发送心跳
func (rf *Raft) heartBeat_paralleled() {
	for i := 0; i < len(rf.peers); i++ {
		if i == rf.me {
			continue
		}

		go rf.sync_appendEntries(i)
	}
}

// The ticker go routine starts a new election if this peer hasn't received
// heartsbeats recently.
func (rf *Raft) ticker() {

	for !rf.killed() {
		// println(strconv.Itoa(rf.me) + " term " + strconv.Itoa(rf.currentTerm))
		// Your code here to check if a leader election should
		// be started and to randomize sleeping time using
		// time.Sleep().

		select {
		case <-rf.actionTimer.C:
			{
				rf.mu.Lock()
				switch rf.role {
				case 0: // leader
					rf.actionTimer.Reset(HEARTBEAT_PERIOD)
					rf.mu.Unlock()
					rf.heartBeat_paralleled()
				case 1: // follower
					rf.role = 2
					rf.votedFor = -1
					rf.actionTimer.Reset(time.Duration(rand.Intn(400)+int(100)) * time.Millisecond)
					rf.persist()
					rf.mu.Unlock()
				case 2: // candidate
					rf.actionTimer.Reset(ELECTION_TIMEOUT_PERIOD + time.Duration(rand.Intn(200))*time.Millisecond)
					rf.currentTerm++
					rf.votedFor = rf.me
					rf.persist()
					rf.mu.Unlock()
					rf.start_vote_paralleled()
				default:
					panic("unkown certification")
				}
			}
		}
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.applyCh = applyCh
	rf.currentTerm = 0
	rf.votedFor = -1
	rf.log = []LogEntry{{0, -100, 0}}
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.role = 1
	rf.actionTimer = *time.NewTimer(0)
	rf.nextIndex = make([]int, len(rf.peers))
	rf.matchIndex = make([]int, len(rf.peers))
	for i := 0; i < len(rf.peers); i++ {
		rf.nextIndex[i] = rf.log[len(rf.log)-1].Index + 1
		rf.matchIndex[i] = 0
	}

	rf.applyLogSignal = make(chan bool, 32)

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()

	// 新开守护线程应用日志
	go rf.applyLogDaemon()

	return rf
}
